package com.xiaoyu.tio.redis.handler;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.client.intf.ClientAioHandler;
import org.tio.core.ChannelContext;
import org.tio.core.GroupContext;
import org.tio.core.exception.AioDecodeException;
import org.tio.core.intf.Packet;
import org.tio.utils.hutool.StrUtil;

import com.alibaba.fastjson.JSON;
import com.xiaoyu.tio.redis.MessageResponse;
import com.xiaoyu.tio.redis.RedisPacket;

public class RedisHandler implements ClientAioHandler {
	
	private final static Logger logger = LoggerFactory.getLogger(RedisHandler.class);

	private static final String CRLF = "\r\n";

	private static final byte CR_BYTE = 13;

	private static final byte LF_BYTE = 10;

	private static final String MULTI_BULK_REPLY = "*";

	private static final byte MULTI_BULK_BYTE = 42;

	private static final String BULK_REPLY = "$";

	private static final byte BULK_BYTE = 36;

	private static final String INT_REPLY = ":";

	private static final byte INT_BYTE = 58;

	private static final String STATUS_REPLY = "+";
	
	private static final byte  STATUS_BYTE = 43;
	
	private static final String ERROR_REPLY = "-";
	
	private static final byte ERROR_BYTE = 45;

	private static final int MIN_LENGTH = 3;

	private final ConcurrentHashMap<Long, CompletableFuture<MessageResponse>> data = new ConcurrentHashMap<>();

	private final ConcurrentLinkedQueue<Long> stack = new ConcurrentLinkedQueue<>();
	
	@Override
	public Packet decode(ByteBuffer buffer, int limit, int position, int readableLength, ChannelContext channelContext)
			throws AioDecodeException {
		if (MIN_LENGTH >= readableLength) {
			return null;
		}
		RedisPacket rp = new RedisPacket();
		byte[] dst = new byte[readableLength];
		byte[] need = null;
		Integer end = null;
		System.arraycopy(buffer.array(), position, dst, 0, readableLength);
		loop: switch (dst[0]) {
		case BULK_BYTE:
			end = getPacketLength(dst, 0, 1);
			if (end == null) {
				return null;
			}
			break loop;
		case MULTI_BULK_BYTE:
			int n = 0;
			for (int z = 1; z < dst.length; z++) {	
				if (dst[z] == CR_BYTE && dst[z + 1] == LF_BYTE) {
					byte[] h = new byte[n];
					for (int p = 1, x = 0; p < z; p++, x++) {
						h[x] = dst[p];
					}
					Integer t = null;
					try {
						t = Integer.valueOf(new String(h));
					}catch (NumberFormatException e) {
						throw new AioDecodeException("协议格式错误！");
					}
					if(t <= 0) {
						end = z + 2;
					}else {
						end = getPacketLength(dst, z + 2, t);
						if (end == null) {
							return null;
						}
					}
					break;
				}else {
					n++;
				}
			}
			break loop;
		case INT_BYTE:
			end = nextCRLFIndex(dst, 0);
			if (end == null) {
				return null;
			}
			end = end + 1;
			break loop;
		case STATUS_BYTE:
			end = nextCRLFIndex(dst, 0);
			if (end == null) {
				return null;
			}
			end = end + 1;
			break loop;
		case ERROR_BYTE:
			end = nextCRLFIndex(dst, 0);
			if (end == null) {
				return null;
			}
			end = end + 1;
			break loop;
		default:
			throw new AioDecodeException("未知redis命令  !!");
		}
		need = new byte[end];
		buffer.get(need);
		rp.setRes(need);
//		rp.setCmd(new String(dst[0]));
		return rp;
	}
	/**
	 * 从指定的下标，从数组中搜索下一个结束符（\r\n）的下标，超过包的长度返回null
	 * @param startIndex
	 * @param dst
	 * @return
	 */
	private Integer nextCRLFIndex(byte[] dst, int startIndex) {
		if(startIndex + 1 >= dst.length) {
			return null;
		}
		for (int z = startIndex; z < dst.length; z++) {
			int lfIndex = z + 1;
			if(lfIndex >= dst.length) {
				return null;
			}
			if (dst[z] == CR_BYTE && dst[lfIndex] == LF_BYTE) {
				return lfIndex;
			}
		}
		return null;
	}

	/**
	 * 从字节数组中获取完整包的长度
	 * @param dst 
	 * @param startIndex
	 * @param parmNum
	 * @return
	 */
	private Integer getPacketLength(byte[] dst, int startIndex, int parmNum) {	
		int i = 0;
		Integer endIndex = null;
		for (int j = startIndex; j < dst.length; j++) {
			int n = 0;
			if (i < parmNum) {
				loop: switch (dst[j]) {
				case BULK_BYTE:
					for (int z = j + 1; z < dst.length; z++) {
						int lfIndex = z + 1;
						if(lfIndex >= dst.length) {
							return null;
						}	
						if (dst[z] == CR_BYTE && dst[lfIndex] == LF_BYTE) {
							byte[] h = new byte[n];
							for (int p = j + 1, x = 0; p < z; p++, x++) {
								h[x] = dst[p];//获取参数长度的值
							}
							Integer t = Integer.valueOf(new String(h));//转换成int类型
							endIndex = t + lfIndex + 2;//得到完整参数的长度
							if (endIndex + 1 > dst.length) {
								return null;
							} else {
								j = endIndex;
								i++;
								break;
							}
						}else {
							n++;
						}
					}
					break loop;
				case INT_BYTE:
					endIndex = nextCRLFIndex(dst, j);
					if(endIndex == null) {
						return null;
					}else {
						j = endIndex;
						i++;
					}				
					break loop;
				case STATUS_BYTE:
					endIndex = nextCRLFIndex(dst, j);
					if(endIndex == null) {
						return null;
					}else {
						j = endIndex;
						i++;
					}				
					break loop;
				case ERROR_BYTE:
					endIndex = nextCRLFIndex(dst, j);
					if(endIndex == null) {
						return null;
					}else {
						j = endIndex;
						i++;
					}				
					break loop;
				}
			} else {
				break;
			}
		}
		if(i != parmNum) {
			return null;
		}
		return endIndex + 1;
	}

	@Override
	public ByteBuffer encode(Packet packet, GroupContext groupContext, ChannelContext channelContext) {
		RedisPacket rp = (RedisPacket) packet;
		String data = null;
		String cmd = rp.getCmd();
		List<String> parms = rp.getParms();
		int parmNum = 1 + parms.size();// 参数数量
		int parmOneLength = cmd.length();// 命令的的字节数
		data = MULTI_BULK_REPLY + parmNum + CRLF + BULK_REPLY + parmOneLength + CRLF + cmd + CRLF;
		for (int i = 0; i < parms.size(); i++) {
			if (StrUtil.isNotBlank(parms.get(i))) {
				data = data + BULK_REPLY + parms.get(i).length() + CRLF + parms.get(i) + CRLF;
			}
		}
		ByteBuffer buffer = ByteBuffer.allocate(data.length());
		buffer.order(groupContext.getByteOrder());
		buffer.put(data.getBytes());
		return buffer;
	}

	@Override
	public void handler(Packet packet, ChannelContext channelContext) throws Exception {
		RedisPacket helloPacket = (RedisPacket) packet;
		byte[] body = helloPacket.getRes();
		if (body != null) {
			String str = new String(body);
			String[] ss = str.split(CRLF);
			if (ss.length > 0) {
				List<Object> data = replyHandler(ss);
				Long id = stack.poll();
				if (id != null) {
					MessageResponse messageResponse = new MessageResponse();
					messageResponse.setData(data);
					messageResponse.setId(id);
					CompletableFuture<MessageResponse> f = this.data.get(id);
					f.complete(messageResponse);
					this.data.remove(id);
				}else {
					logger.warn("未找到消息ID:{} 数据:{}", id, JSON.toJSONString(data));
				}	
			} else {
				logger.error("数据格式错误:{}" + str);
			}
		}
	}

	@Override
	public Packet heartbeatPacket() {
		return null;
	}

	private List<Object> replyHandler(String[] cmd) {
		List<Object> result = new ArrayList<Object>();
		for (int i = 0; i < cmd.length; i++) {
			String cs = cmd[i].substring(0, 1);
			String v = cmd[i].substring(1);
			loop: switch (cs) {
			case BULK_REPLY:
				Integer dataLength = Integer.valueOf(v);
				if (dataLength == -1) {
					result.add(null);
				}
				break loop;
			case INT_REPLY:
				Integer number = Integer.valueOf(v);
				result.add(number);
				break loop;
			case ERROR_REPLY:
				result.add(v);
				break loop;
			case STATUS_REPLY:
				result.add(true);
				break loop;
			case MULTI_BULK_REPLY:

				break loop;
			default:
				result.add(cmd[i]);
				break loop;
			}
		}
		return result;
	}

	public Object getResData(Long id) {
		return data.get(id);
	}
	
	public void stackAdd(Long id) {
		stack.add(id);
	}
	
	public void addMessage(Long id, CompletableFuture<MessageResponse> messageResponse) {
		data.put(id, messageResponse);
	}
	

}
